package com.fs.uc.nsq;

import com.fs.uc.base.MyConfig;
import com.fs.uc.base.domain.Org;
import com.fs.uc.base.domain.Person;
import com.fs.uc.base.repo.OrgRepository;
import com.fs.uc.core.behavior.ParkBehavior;
import com.fs.uc.core.behavior.PersonBehavior;
import com.fs.uc.nsq.data.AbsData;
import com.fs.uc.nsq.data.Staff;
import com.fs.uc.nsq.data.TenantStaff;
import com.fs.uc.nsq.dto.MessageHeader;
import com.fs.uc.nsq.dto.MessageReply;
import com.fs.uc.nsq.dto.OurMessage;
import com.fs.uc.nsq.dto.ReplyHeader;
import com.google.gson.Gson;
import com.sproutsocial.nsq.Publisher;
import com.sproutsocial.nsq.Subscriber;
import io.quarkus.runtime.Startup;
import io.smallrye.common.constraint.Assert;
import io.smallrye.common.constraint.Nullable;
import io.smallrye.mutiny.tuples.Tuple2;

import javax.annotation.PostConstruct;
import javax.enterprise.context.ApplicationScoped;
import java.util.List;
import java.util.Random;

/**
 * @author 崔超
 * 2021/12/8
 */
@Startup
@ApplicationScoped
public class NsqSubscribe {
    private final static String CREATE_STAFF_TOPIC = "BasicInfo-AddStaff";
    private final static String DELETE_STAFF_TOPIC = "BasicInfo-DelStaff";
    private final static String CREATE_CUSTOM_USER_TOPIC = "BasicInfo-AddCustomUser";
    private final static String DELETE_CUSTOM_USER_TOPIC = "BasicInfo-DelCustomUser";
    private final static String NSQ_CHANNEL = "UserCenter";

    private final Subscriber subscriber;
    private final List<String> nsqLookUps;
    private final PersonBehavior personBehavior;
    private final OrgRepository orgRepository;
    private final ParkBehavior parkBehavior;

    public NsqSubscribe(MyConfig config, PersonBehavior personBehavior, ParkBehavior parkBehavior, OrgRepository orgRepository) {
        subscriber = new Subscriber(config.nsq().lookUp());
        nsqLookUps = config.nsq().nodes();
        this.orgRepository = orgRepository;
        this.parkBehavior = parkBehavior;
        this.personBehavior = personBehavior;
    }

    @PostConstruct
    void subscribe() {
        subscriber.subscribe(CREATE_STAFF_TOPIC, NSQ_CHANNEL, this::addEnterpriseStaff);
        subscriber.subscribe(DELETE_STAFF_TOPIC, NSQ_CHANNEL, this::delEnterpriseStaff);
        subscriber.subscribe(CREATE_CUSTOM_USER_TOPIC, NSQ_CHANNEL, this::addTenantStaff);
        subscriber.subscribe(DELETE_CUSTOM_USER_TOPIC, NSQ_CHANNEL, this::delTenantStaff);
    }

    void addEnterpriseStaff(byte[] data) {
        Tuple2<MessageHeader, Staff> tuple = extractData(data, Staff.class);
        MessageHeader msgHeader = tuple.getItem1();
        Staff staff = tuple.getItem2();

        if (staff.getPhone() == null || staff.getEnterpriseId() == null) {
            String errMsg = "nsq消息获取到的员工信息不完整";
            replyFail(msgHeader.getReplyTopic(), msgHeader.getObjectId(), errMsg);
            return;
        }

        Org enterprise = orgRepository.findByRemoteId(staff.getEnterpriseId());
        Assert.assertNotNull(enterprise);
        Person person = personBehavior.createByOrgAdmin(staff.getPhone(), staff.getName());
        personBehavior.joinOrg(person.id, enterprise.id);

        replySuc(msgHeader.getReplyTopic(), msgHeader.getObjectId(), person.id.toString());
    }

    void addTenantStaff(byte[] data) {
        Tuple2<MessageHeader, TenantStaff> tuple = extractData(data, TenantStaff.class);
        MessageHeader msgHeader = tuple.getItem1();
        TenantStaff tenantStaff = tuple.getItem2();

        if (tenantStaff.getPhone() == null || tenantStaff.getTenantId() == null) {
            String errMsg = "nsq消息获取到的员工信息不完整";
            replyFail(msgHeader.getReplyTopic(), msgHeader.getObjectId(), errMsg);
            return;
        }

        Person person = personBehavior.createByOrgAdmin(tenantStaff.getPhone(), tenantStaff.getName());
        personBehavior.joinOrg(person.id,tenantStaff.getTenantId());

        replySuc(msgHeader.getReplyTopic(), msgHeader.getObjectId(), person.id.toString());
    }

    void delEnterpriseStaff(byte[] data) {
        Tuple2<MessageHeader, Staff> tuple = extractData(data, Staff.class);
        MessageHeader msgHeader = tuple.getItem1();
        Staff staff = tuple.getItem2();

        if (staff.getEnterpriseId() == null || staff.getUserId() == null) {
            String errMsg = "nsq消息获取到的员工信息不完整";
            replyFail(msgHeader.getReplyTopic(), msgHeader.getObjectId(), errMsg);
            return;
        }

        Org org = orgRepository.findByRemoteId(staff.getEnterpriseId());
        Assert.assertNotNull(org);
        personBehavior.leaveOrg(staff.getUserId(),org.id);

        replySuc(msgHeader.getReplyTopic(), msgHeader.getObjectId(), "");
    }

    void delTenantStaff(byte[] data) {
        Tuple2<MessageHeader, TenantStaff> tuple = extractData(data, TenantStaff.class);
        MessageHeader msgHeader = tuple.getItem1();
        TenantStaff tenantStaff = tuple.getItem2();

        if (tenantStaff.getUserId() == null || tenantStaff.getTenantId() == null) {
            String errMsg = "nsq消息获取到的员工信息不完整";
            replyFail(msgHeader.getReplyTopic(), msgHeader.getObjectId(), errMsg);
            return;
        }

        personBehavior.leaveOrg(tenantStaff.getUserId(), tenantStaff.getTenantId());

        replySuc(msgHeader.getReplyTopic(), msgHeader.getObjectId(), "");
    }

    private <T extends AbsData> Tuple2<MessageHeader, T> extractData(byte[] data, @Nullable Class<T> dataClazz) {
        Gson gson = new Gson();
        String strData = new String(data);
        OurMessage message = gson.fromJson(strData, OurMessage.class);
        T dataInstance = gson.fromJson(message.getData(), dataClazz);

        String replyTopic = message.getHeader().getReplyTopic();
        if (replyTopic == null) {
            throw new RuntimeException("nsq消息获取到的replyTopic为空");
        }
        return Tuple2.of(message.getHeader(), dataInstance);
    }

    private void replySuc(String replyTopic, String objectId, String data) {
        nsqReply(replyTopic, objectId, true, "", data);
    }

    private void replyFail(String replyTopic, String objectId, String errMsg) {
        nsqReply(replyTopic, objectId, false, errMsg, "");
    }

    private void nsqReply(String replyTopic, String objectId, Boolean suc, String errMsg, String data) {
        ReplyHeader replyHeader = new ReplyHeader(objectId, NSQ_CHANNEL, suc, errMsg);
        MessageReply reply = new MessageReply(replyHeader, data);
        String strReply = new Gson().toJson(reply);
        int i = new Random().nextInt(nsqLookUps.size());
        Publisher publisher = new Publisher(nsqLookUps.get(i));
        publisher.publish(replyTopic, strReply.getBytes());
        publisher.stop();
    }
}
